package com.atguigu.Flink.state;

import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.function.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.Stream;

// 通过窗口收集指定时间内的所有的数据(不做keyBy) ,
//         *  通过处理函数，统一对窗口内的数据进行聚合处理。
public class Flink13_TopN_ProcessAllWindowFunction {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> ds = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event event, long l) {
                                        return event.getTs();
                                    }
                                })
                );
        ds.print("Input");

        ds.windowAll(
                TumblingEventTimeWindows.of(Time.seconds(10))
        ).process(
                new ProcessAllWindowFunction<Event, String, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<Event, String, TimeWindow>.Context context, Iterable<Event> iterable, Collector<String> collector) throws Exception {
                        //先统计每个url的点击次数
                        HashMap<String, Long> urlViewMap = new HashMap<>();
                        for (Event element : iterable) {
                            Long count = urlViewMap.getOrDefault(element.getUrl(), 0L);
                            urlViewMap.put(element.getUrl(), count + 1);
                        }
                        // 将Map转换成List
                        ArrayList<Tuple2<String, Long>> urlViewList = new ArrayList<>();
                        for (Map.Entry<String, Long> entry : urlViewMap.entrySet()) {
                            urlViewList.add(Tuple2.of(entry.getKey(), entry.getValue()));
                        }
                        //排序
                        urlViewList.sort(
                                new Comparator<Tuple2<String, Long>>() {
                                    @Override
                                    public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
                                        return Long.compare(o2.f1, o1.f1);
                                    }
                                }
                        );
                        //获取窗口信息
                        long start = context.window().getStart();
                        long end = context.window().getEnd();
                        //取Top2
                        StringBuilder resultBuilder = new StringBuilder("==================Top2==================\n");
                        resultBuilder.append("====[ " + start + " ~ " + end + " )====\n") ;
                        for (int i = 0; i < Math.min( 2  , urlViewList.size()); i++) {
                            Tuple2<String, Long> urlViewCount = urlViewList.get(i);
                            resultBuilder.append("NO." + (i+1)  + " URL: " + urlViewCount.f0 + " , 点击次数: " + urlViewCount.f1 + "\n") ;
                        }
                        resultBuilder.append("========================================\n") ;


                        collector.collect(resultBuilder.toString());
                    }
                }
        ).print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
